# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: jaewon, inkyu

import logging
import os
import socket

from absl import app, flags
from tornado.ioloop import IOLoop

from coin.base.timestamp import get_timestamp
from coin.util.queue.sync_publisher import KafkaPublisher
from coin.util.log_watcher.stderr_log_watcher import StderrLogWatcher


class KafkaPublishingHandler:
  def __init__(self, *, hostname, kafka_publisher):
    self._hostname = hostname
    self._kafka_publisher = kafka_publisher
    self._topic_names = {}

  def _topic_name(self, path):
    if path not in self._topic_names:
      relpath = os.path.relpath(path)
      self._topic_names[path] = '%s_%s' % (self._hostname, relpath.replace('/', '.'))
    return self._topic_names[path]

  def on_log(self, path, data):
    self._kafka_publisher.send_message(self._topic_name(path),
                                       data.encode('utf-8'),
                                       timestamp=get_timestamp())


def main(argv):
  FLAGS = flags.FLAGS
  ioloop = IOLoop.current()

  kafka_publisher = KafkaPublisher(kafka_servers=FLAGS.kafka_servers.split(','))
  kafka_publisher.start(ioloop)

  handler = KafkaPublishingHandler(hostname=socket.gethostname(), kafka_publisher=kafka_publisher)

  watcher = StderrLogWatcher(target_dir=argv[1], log_callback=handler.on_log)

  try:
    watcher.start(ioloop)
    ioloop.start()
  except (KeyboardInterrupt, SystemExit):
    return


if __name__ == "__main__":
  logging.basicConfig(
      level=logging.DEBUG,
      format='%(levelname)8s %(asctime)s %(name)s %(filename)s:%(lineno)d] %(message)s')

  flags.DEFINE_string('kafka_servers',
                      'coin-kafka.corp.prestolabs.io:9092',
                      'Kafka server list separated by comma.')

  app.run(main)
